package com.study.project.config.kafka;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

/**
 * @program: studyProject
 * @see:
 * @author: yangjie
 * @create: 2024-04-13 15:48
 **/
@Component
public class KafkaConsumer {
    /*
     * public enum AckMode {
     *     // 当每一条记录被消费者监听器（ListenerConsumer）处理之后提交
     *     RECORD,
     *     // 当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后提交
     *     BATCH,
     *     // 当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后，距离上次提交时间大于TIME时提交
     *     TIME,
     *     // 当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后，被处理record数量大于等于COUNT时提交
     *     COUNT,
     *     // TIME |　COUNT　有一个条件满足时提交
     *     COUNT_TIME,
     *     // 当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后, 手动调用Acknowledgment.acknowledge()后提交
     *     MANUAL,
     *     // 手动调用Acknowledgment.acknowledge()后立即提交
     *     MANUAL_IMMEDIATE,
     * } 
     */

    /**
     * 监听消费 groupId相同的情况下，消费相同的topic，按照先后顺序，第一个生效去消费,其余相同的不会去消费
     * 
     * @param record 记录
     */
    @KafkaListener(topics = {KafkaTopic.TEST}, groupId = KafkaGroupId.TEST1)
    public void onNormalMessage(ConsumerRecord<String, Object> record) {
        System.out.println("简单消费：" + record.topic() + "-" + record.partition() + "=" + record.value());
    }

    /**
     * 监听消费
     *
     * @param record 记录
     */
    @KafkaListener(topics = {KafkaTopic.TEST}, groupId = KafkaGroupId.TEST2)
    public void onNormalGroupIdMessage(ConsumerRecord<String, Object> record) {
        System.out.println("groupId简单消费：" + record.topic() + "-" + record.partition() + "=" + record.value());
    }

}